/** @Date 下午5:06:23
  * @version 1.0.0
  * @author shangyd
  * Copyright (c) 2015
  */
package com.sdw.soft.activemq;

import java.io.Serializable;
import java.util.Map;
import java.util.Queue;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JMSMessageRunner implements Runnable{

	private static final Logger logger = LoggerFactory.getLogger(JMSMessageRunner.class);
	
	private String brokerUrl;
	private static final Long THREAD_INTERVAL = new Long(10 * 1000);
	private ConnectionFactory connectionFactory ;
	private Connection connection;
	private Session session;
	private Destination destination;
	private MessageProducer messageProducer;
	
	public void setBrokerUrl(String brokerUrl) {
		this.brokerUrl = brokerUrl;
	}

	public JMSMessageRunner (String brokerUrl){
		this.brokerUrl = brokerUrl;
	}
	
	@Override
	public void run() {
		while(true){
			JMSMessageSender jmsSender = JMSMessageSender.instance();
			if(!jmsSender.getQueues().isEmpty()){
				try {
					connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD,brokerUrl);
					connection = connectionFactory.createConnection();
					connection.start();
					session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
					Map<String,Queue> copyContent = jmsSender.copyAndClearMap();
					Set<String> queueNames = copyContent.keySet();
					for(String queueName : queueNames){
						destination = session.createQueue(queueName);
						messageProducer = session.createProducer(destination);
						messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
						Queue<Map> contents = copyContent.get(queueName);
						for(Map content : contents){
							ObjectMessage message = session.createObjectMessage();
							if(content != null){
								Set<String> params = content.keySet();
								for(String param : params){
									Object value = content.get(param);
									if(value instanceof String){
										message.setStringProperty(param, (String)value);
									}else if(value instanceof Serializable){
										message.setObject((Serializable)value);
									}
								}
							}
							messageProducer.send(message);
							session.commit();
							logger.info("--------------send message successsfully-------------------");
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
					logger.error(e.getMessage(),e);
				}
				
			}
			
			try {
				Thread.sleep(THREAD_INTERVAL);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
